package com.shuai.frame.registry;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.shuai.frame.config.XsRPCRegistryCenterConfig;
import com.shuai.frame.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;


@Slf4j
public class ETCDRegistryCenter implements RegistryCenter {
    private Client client;
    private KV kvClient;

    /**
     * 服务节点本地缓存
     */
    private final RegistryLocalCache localCache = new RegistryLocalCache();

    /**
     * 服务节点本地监听key值的缓存
     */
    private final Set<String> watchingKeys = new ConcurrentHashSet<>();

    /**
     * 本地注册节点的key集合
     */
    private final Set<String> localCacheRegistryKeys = new HashSet<>();
    /**
     * 根节点
     */
    private static final String ROOT_PATH = "/xs-rpc/";

    @Override
    public void init(XsRPCRegistryCenterConfig config) {
        client = Client.builder().endpoints(config.getRegistryAddress())
                .connectTimeout(Duration.ofMillis(config.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        // 开启心跳检测
        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws ExecutionException, InterruptedException {
        Lease leaseClient = client.getLeaseClient();
        long id = leaseClient.grant(30).get().getID();
        String registryKey = ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registryKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);
        // 将键值与租约关联，30秒后过期
        kvClient.put(key, value, PutOption.builder().withLeaseId(id).build()).get();
        // 讲服务节点的key值添加到本地缓存中
        localCacheRegistryKeys.add(registryKey);
        log.info("服务节点:{} 注册成功!", registryKey);
    }

    @Override
    public void unregister(ServiceMetaInfo serviceMetaInfo) throws ExecutionException, InterruptedException {
        kvClient.delete(ByteSequence.from(ROOT_PATH + serviceMetaInfo.getServiceNodeKey(), StandardCharsets.UTF_8)).get();
        // 从本地缓存中移除服务节点
        localCacheRegistryKeys.remove(ROOT_PATH + serviceMetaInfo.getServiceNodeKey());
        log.info("服务节点:{} 注销成功!", ROOT_PATH + serviceMetaInfo.getServiceNodeKey());
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscover(String serviceKey) throws ExecutionException, InterruptedException {
        // 先从本地缓存中获取
        if (CollectionUtil.isNotEmpty(localCache.getLocalCache())) {
            log.info("从本地缓存中进行服务: {} 的服务发现", serviceKey);
            return localCache.getLocalCache();
        }
        log.info("从etcd中进行服务: {} 的服务发现", serviceKey);
        String searchPath = ROOT_PATH + serviceKey + "/";
        GetOption getOption = GetOption.builder().isPrefix(true).build();
        List<KeyValue> kvs = kvClient.get(ByteSequence.from(searchPath, StandardCharsets.UTF_8), getOption)
                .get()
                .getKvs();
        List<ServiceMetaInfo> serviceMetaInfos = kvs.stream().map(item -> {
            String key = item.getKey().toString(StandardCharsets.UTF_8);
            watch(key);
            String value = item.getValue().toString(StandardCharsets.UTF_8);
            return JSONUtil.toBean(value, ServiceMetaInfo.class);
        }).collect(Collectors.toList());
        // 设置服务到本地服务缓存中
        localCache.setLocalCache(serviceMetaInfos);
        return serviceMetaInfos;
    }

    @Override
    public void destroyService() {
        for (String registryKey : localCacheRegistryKeys) {
            try {
                kvClient.delete(ByteSequence.from(registryKey, StandardCharsets.UTF_8)).get();
                log.info("节点:{}  下线成功!", registryKey);
            } catch (Exception e) {
                log.error("节点:{}下线异常: {}", registryKey, e.getMessage());
            }
        }
        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
        log.info("节点下线");
    }


    /**
     * 心跳检测
     */
    @Override
    public void heartBeat() {
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                for (String registryKey : localCacheRegistryKeys) {
                    try {
                        List<KeyValue> kvs = kvClient.get(ByteSequence.from(registryKey, StandardCharsets.UTF_8))
                                .get().getKvs();
                        // 节点已过期
                        if (CollectionUtil.isEmpty(kvs)) {
                            continue;
                        }
                        // 续签节点（重新注册）
                        KeyValue keyValue = kvs.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                        log.info("节点:{} 续约成功!", registryKey);
                    } catch (Exception e) {
                        log.error("续签失败: {}", e.getMessage());
                        throw new RuntimeException("续签失败: " + e.getMessage());
                    }
                }
            }
        });
        // 支持秒级定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(String serviceKey) {
        Watch watchClient = client.getWatchClient();
        if (watchingKeys.add(serviceKey)) {
            watchClient.watch(ByteSequence.from(serviceKey, StandardCharsets.UTF_8), watchResponse -> {
                for (WatchEvent event : watchResponse.getEvents()) {
                    if (event.getEventType() == WatchEvent.EventType.DELETE) {// 清理本地服务缓存
                        log.info("服务节点:{} 删除, 清理本地服务缓存", serviceKey);
                        localCache.clear();
                    }
                }
            });
        }
    }
}
